package com.daidai.table;

import com.daidai.sink.domain.User;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

public class DataStreamtoTable {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        DataStreamSource<User> source = env.fromCollection(Arrays.asList(
                new User(1, "user1", "password1"),
                new User(2, "user2", "password2"),
                new User(3, "user3", "password3")
        ));

        Table user = tableEnvironment.fromDataStream(source, $("id"), $("username"), $("password"));

        Table table = user.select($("id"), $("username"), $("password"));

        DataStream<User> appendStream = tableEnvironment.toAppendStream(table, TypeInformation.of(User.class));

        appendStream.print();

        env.execute();
    }
}
